package com.cnlive.logpare.play

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import com.cnlive.logpare.util.Constant
import scala.collection.mutable.HashMap
import com.cnlive.logpare.util.IPSeekerExt
import scala.util.parsing.json.JSON
import java.text.SimpleDateFormat
import java.util.Locale
import java.util.Properties
import java.sql.DriverManager
import java.sql.PreparedStatement
import java.sql.Connection

object ParsePlayLog {
    val dateFormat=new SimpleDateFormat("yyyyMMddHHmmss",Locale.ENGLISH)
    val dateFormat2=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss",Locale.ENGLISH)
    def main(args: Array[String]): Unit = {
       var env=1 //运行环境 1.本地环境，2，本地测试环境.3远程测试环境
       if(args.length>0){
         env=args(0).toInt 
       }
       val files=env match{
            case 1 => "file:///F:/工作资料/133_8081_openapi_access_201706140030"
            case _ => "hdfs://sxt/nginx_logs/app/offline/2017-06-15/*"
       }
       val qqwryPath=env match{
                case 2 =>Constant.qqwryLocalServer
                case 3 =>Constant.qqwryServer
                case _=>""
       }
        val zk_addr=env match{
              case 3=>"jdbc:phoenix:node001,node002,node003:2181"
              case _=>"jdbc:phoenix:node180,node181,node182:2181"
        }
        val conf = new SparkConf().setAppName("ParsePlayLog")
        if(env==1){
          conf.setMaster("local")
        }
        val listApid=List("9f0ccea1-716f-4322-9d76-82e4fb5727a4" //liveplayByH5
                ,"0f944919-039f-4925-bc56-f778fe4a7b18"//liveplayBySWF
                ,"c77c1d71-6402-4321-a371-bac3717ef8ba"//liveplayByAPP
                ,"183ff68f-fa6d-4d4a-ad14-fe17cab03a88"//vodplayBySWF
                ,"cba8393a-699e-4fbf-b655-b854d6a3b43d"//vodplayByH5
                ,"dfbba809-6d7d-41c6-bd6a-da732946f513")//vodplayByAPP
        val listPlayStr=List("liveplayByH5",
               "liveplayBySWF",
               "liveplayByAPP",
               "vodplayBySWF",
               "vodplayByH5",
               "vodplayByAPP")
        val sc = new SparkContext(conf)
        sc.textFile(files).mapPartitions(x=>{
           val iPSeekerExt=env match {
              case 1 =>  IPSeekerExt.getInit()
              case 2 =>  IPSeekerExt.getInit(true,qqwryPath)
              case 3 =>  IPSeekerExt.getInit(true,qqwryPath)
           }
           val props=new Properties();
		       val conn = DriverManager.getConnection(zk_addr, props);
		       val sql="""
    		    upsert into playlog 
    		    (NGINXTIME,PLAYTYPE,PLAT,SPID, APPID,IP,COUNTRY,CITY,LOCATION,OPERATOR,HASHCODE) 
    		     values(?,?,?,?,?,?,?,?,?,?,?)
    		   """
		       val prest = conn.prepareStatement(sql)
		       val sessionMap=new HashMap[String, Int]()
            sessionMap.put("count", 0)
           var result = List[String]()
           var count=0
           while (x.hasNext) {
              val lineText=x.next()
              for((str,index) <-listPlayStr.zipWithIndex){
                if(lineText.indexOf(str) != -1){
                   if(x.hasNext){
                     x.next()
                     if(x.hasNext){
                       val linexText3=x.next()
                       if(linexText3.indexOf(listApid(index)) != -1){
                          parLineText(iPSeekerExt,lineText,linexText3,prest,conn,sessionMap)//对api数据进行解析
                       }
                     }
                   }
                }
              }
           }
		       conn.commit()
           conn.close()
           result.iterator
        }, true).count()
    }
    def parLineText(iPSeekerExt:IPSeekerExt,lineText:String,line3Text:String,prest:PreparedStatement,con:Connection,sessionMap:HashMap[String, Int]){
      val mapRequest=new HashMap[String, String]()
      val lineArr= lineText.split("\\|")
      if(lineArr.length==2){
          lineArr(1).split(",").foreach(x=>{
                val sk=x.split(":")
                if(sk.length==2){
                    mapRequest.put(sk(0), sk(1))
                }
          })
      }
      val line3Arr=line3Text.split("\\|")
      if(line3Arr.length ==3){
          val time=line3Arr(0).split("INFO")(0).replaceAll("\\[openapi\\] ", "")
          val nginxtime=dateFormat.format(dateFormat2.parse(time)).toLong
          val json=JSON.parseFull(line3Arr(2).replaceAll("API_JSON_DATA:", ""))
          val mapa=json  match {
            case Some(x:Map[String,String]) =>x
            case _ =>  Map[String,String]()
          }
          val apiType=mapa.getOrElse("api_id", "") match{
            case "9f0ccea1-716f-4322-9d76-82e4fb5727a4" =>(1,303)//("live","H5")
            case "0f944919-039f-4925-bc56-f778fe4a7b18" =>(1,301)//("live","SWF")
            case "c77c1d71-6402-4321-a371-bac3717ef8ba" =>(1,302)//("live","APP")
            case "183ff68f-fa6d-4d4a-ad14-fe17cab03a88" =>(2,303)//("vod","H5")
            case "cba8393a-699e-4fbf-b655-b854d6a3b43d" =>(2,301)//("vod","SWF")
            case "dfbba809-6d7d-41c6-bd6a-da732946f513" =>(2,302)//("vod","APP")
            case _=>(0,399)//("","")
          }
          val appid=mapa.getOrElse("app_id", "")
          val spid=mapa.getOrElse("sp_id", "")//节目id
          val ip=mapRequest.getOrElse("ip", "127.0.0.1")
          val d=iPSeekerExt.analyticIp(ip)
          val city=d.getCity
          val province=d.getProvince
          val country=d.getCountry
          val operator=d.getOperator
          var index=1
          //NGINXTIME,PLAYTYPE,PLAT,SPID, APPID,IP,COUNTRY,CITY,LOCATION,OPERATOR,HASHCODE
          prest.setLong(index, nginxtime);index=index+1
          prest.setInt(index, apiType._1);index=index+1
          prest.setInt(index, apiType._2);index=index+1
          prest.setString(index, spid);index=index+1
          prest.setString(index, appid);index=index+1
          prest.setString(index, ip);index=index+1
          prest.setInt(index, country);index=index+1
          prest.setString(index, city);index=index+1
          prest.setString(index, province);index=index+1
          prest.setInt(index, operator);index=index+1
          prest.setLong(index, (lineText+line3Text).hashCode());
          prest.execute()
          con.commit()
       }
    }
    
}